package com.offcn.bigdata.flink.watermark;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Arrays;
import java.util.List;

public class TumblingEventWindowExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> socketStream = env.socketTextStream("bigdata01", 9999);

        DataStream<Tuple2<String, Long>> resultStream = socketStream
            // Time.seconds(3)有序的情况修改为0
            .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {
                @Override
                public long extractTimestamp(String element) {
                    long eventTime = Long.parseLong(element.split(" ")[0]);
                    System.out.println(eventTime);
                    return eventTime;
                }
            })
            .map(new MapFunction<String, Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> map(String value) throws Exception {
                    return Tuple2.of(value.split(" ")[1], 1L);
                }
            })
            .keyBy(0)
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            .reduce(new ReduceFunction<Tuple2<String, Long>>() {
                @Override
                public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                    return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
                }
            });
        resultStream.print();
        env.execute();
    }

    public static List<String> getOrderList() {
        return Arrays.asList(
                "10000 a",
                "11000 a",
                "12000 b",
                "13000 b",
                "14000 a",
                "19888 a",
                "13000 a",
                "20000 a",
                "11000 a",
                "12000 b",
                "21000 b",
                "22000 a",
                "29999 a"
        );
    }
    public static List<String> getOrdernessList() {
        return Arrays.asList(
                "10000 a",
                "11000 a",
                "12000 b",
                "20000 a",
                "21000 a",
                "22000 b",
                "23000 a",
                "24000 a",
                "29000 b",
                "30000 a",
                "22000 a",
                "23000 a",
                "33000 a"
        );
    }
}